package cn._51doit.live.udfs;

import cn._51doit.live.pojo.ItemEventCount;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

public class HotGoodsWindowFunction implements WindowFunction<Integer, ItemEventCount, Tuple3<String, String, String>, TimeWindow> {

    @Override
    public void apply(Tuple3<String, String, String> tp, TimeWindow window, Iterable<Integer> input, Collector<ItemEventCount> out) throws Exception {
        String cid = tp.f0;
        String pid = tp.f1;
        String eid = tp.f2;
        Integer count = input.iterator().next();
        long start = window.getStart();
        long end = window.getEnd();
        out.collect(new ItemEventCount(cid, pid, eid, count, start, end));
    }
}
